package wordcountdemo

import org.apache.flink.api.java.utils.ParameterTool
import org.apache.flink.streaming.api.scala._

/**
  * 1、yarn-session模式
  * yarn-session.sh -n 2 -s 2 -jm 1024 -tm 1024 -nm test -d
  * 其中： -n(--container)：TaskManager 的数量。
  *        -s(--slots)： 每个 TaskManager 的 slot 数量，默认一个 slot 一个 core，默认每个 taskmanager 的 slot 的个数为 1，有时可以多一些 taskmanager，做冗余。
  *        -jm：JobManager 的内存（单位 MB)。
  *        -tm：每个 taskmanager 的内存（单位 MB)。
  *        -nm：yarn 的 appName(现在 yarn 的 ui 上的名字)。
  *        -d：后台执行。
  * flink run -c wordcountdemo.StreamWordCount –p 2 flinkWordCount.jar --host master –port 9999
  *
  *
  * 2、Per Job Cluster模式 不用启动flink
  * flink run –m yarn-cluster -c wordcountdemo.StreamWordCount –p 2 flinkWordCount.jar --host master –port 9999
  */
object StreamWordCount {
  def main(args: Array[String]): Unit = {
    //从外部命令中提取参数
    val tool = ParameterTool.fromArgs(args)
    val host: String = tool.get("host")
    val port: Int = tool.getInt("port")

    //，getExecutionEnvironment
    // 会根据查询运行的方式决定返回什么样的运行环境，是最常用的一种创建执行环境的方式。
    // flink会自动的给我们做判断 createLocalEnvironment（本地）或createRemoteEnvironment（集群）
    val streamEnv = StreamExecutionEnvironment.getExecutionEnvironment
    //根据运行环境来设置并行度，这里是全局设置，默认并行度是CPU的核心数量来的
    //    streamEnv.setParallelism(10)
    //并行度优先级 先看代码，代码有个别算子设置了并行度，这是没法变的，没有配的看全局的并行度是几，
    // 如果没配看提交任务时的并行度，如果还没有配，最后看集群的配置文件

    //全局禁止合并任务链
//    streamEnv.disableOperatorChaining()

    //当前流的并行度 一般指的是算子中最大的并行度，与slot有关
    //slot并没有对CPU做隔离，只是对内存做独享的隔离
    val WordCounts: DataStream[String] = streamEnv.socketTextStream(host, port)
      .slotSharingGroup("a") //slotSharingGroup设置一个共享组a另一个slot

    // flatMap 和 Map 需要引用的隐式转换
    //import org.apache.flink.streaming.api.scala._
    //流处理没有groupBy ，但是有keyBy，keyBy是基于hashCode做的
    val resultWordCounts: DataStream[(String, Int)] = WordCounts.flatMap(s => {
      s.split(" ")
    })
      .map((_, 1)) .startNewChain()
      .slotSharingGroup("b")//startNewChain 从当前算子开始一个新的任务链
      .keyBy(0)
      .sum(1).slotSharingGroup("a")

    //每个操作，针对每一个算子都可以进行设置并行度，可以手动设置并行度
    //每个单词的执行CPU值通过哈希值来做的，原因是上面进行了分组
    resultWordCounts.print()


    streamEnv.execute("Socket stream word count")
  }
}
